use futures_util::{SinkExt, StreamExt};
use http_body_util::Full;
use hyper::body::Bytes;
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper::{Request, Response};
use std::convert::Infallible;
use std::net::SocketAddr;
use tokio::net::{TcpListener, TcpStream};
use tokio_tungstenite::{accept_async, tungstenite::Error};
use tracing::{debug, info, instrument};
use tracing_subscriber::{fmt, layer::SubscriberExt, util::SubscriberInitExt};
use tungstenite::Result;

#[instrument]
fn trace_me(a: u32, b: u32) -> u32 {
    info!("aaaaaaaaaaa");
    debug!("bbbbbbbbbb");
    a + b
}

async fn hello(_: Request<hyper::body::Incoming>) -> Result<Response<Full<Bytes>>, Infallible> {
    Ok(Response::new(Full::new(Bytes::from("Hello, World!"))))
}

async fn accept_connection(peer: SocketAddr, stream: TcpStream) {
    if let Err(e) = handle_connection(peer, stream).await {
        match e {
            Error::ConnectionClosed | Error::Protocol(_) | Error::Utf8 => (),
            err => eprintln!("Error processing connection: {}", err),
        }
    }
}

async fn handle_connection(peer: SocketAddr, stream: TcpStream) -> Result<()> {
    let mut ws_stream = accept_async(stream).await.expect("Failed to accept");

    println!("New WebSocket connection: {}", peer);

    while let Some(msg) = ws_stream.next().await {
        let msg = msg?;
        if msg.is_text() || msg.is_binary() {
            ws_stream.send(msg).await?;
        }
    }

    Ok(())
}

#[tokio::main]
async fn main() {
    tracing_subscriber::registry().with(fmt::layer()).init();

    //console_subscriber::init();

    let listener2 = TcpListener::bind("localhost:3465").await.unwrap();

    let listener = TcpListener::bind("127.0.0.1:9002").await.unwrap();

    let listener3 = TcpListener::bind("127.0.0.1:9003").await.unwrap();

    loop {
        tokio::select! {
            _ = async {
                while let Ok((_stream, _)) = listener2.accept().await {
                    tokio::spawn(async move { println!("1111 listener") });
                    tracing::info!("111111111111111111");
                    tracing::warn_span!("222222222222222222");
                    println!("{}", trace_me(2, 3));
                }
            } => {}
            _ = async {
                while let Ok((stream, _)) = listener.accept().await {
                    let peer = stream.peer_addr().expect("connected streams should have a peer address");
                    println!("Peer address: {}", peer);
                    tokio::spawn(accept_connection(peer, stream));
                }
            } => {}
            _ = async {
                while let Ok((stream, _)) = listener3.accept().await {
                    tokio::spawn(async move {
                        if let Err(err) = http1::Builder::new()
                        .serve_connection(stream, service_fn(hello))
                        .await
                        {
                            println!("Error serving connection: {:?}", err);
                        }
                    });
                }
            } => {}
        }
    }
}
